package com.nd.flink;

import com.nd.flink.cli.CliFrontendParser;
import com.nd.flink.entity.ParamsInfo;
import com.nd.flink.entity.ResultInfo;
import com.nd.flink.enums.ETaskStatus;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Date: 2021/10/1
 *
 * @author todd5167
 */
public class CliSubmitterMain {
    private static final Logger LOG = LoggerFactory.getLogger(CliSubmitterMain.class);

    public static void main(String[] args) throws Exception {
        ParamsInfo paramsInfo = new CliFrontendParser().parseParamsInfo(args);
        LOG.info("parsed params info:{}", paramsInfo);

        ClusterClient clusterClient = ClusterClient.INSTANCE;

        ResultInfo jobResult =
                paramsInfo.isOpenSecurity()
                        ? clusterClient.submitFlinkJobWithKerberos(paramsInfo)
                        : clusterClient.submitFlinkJob(paramsInfo);

        LOG.info(
                "------------------------------------submit result--------------------------------------------");
        LOG.info(
                String.format(
                        "job submit result, appId:%s, jobId:%s, msg:%s",
                        jobResult.getAppId(), jobResult.getJobId(), jobResult.getMsg()));
        LOG.info(
                "---------------------------------------------------------------------------------------------");

        paramsInfo.setApplicationId(jobResult.getAppId());

        ETaskStatus eTaskStatus;
        while (true) {
            eTaskStatus = clusterClient.getYarnJobStatus(paramsInfo);
            LOG.info(
                    "------------------------------------current job status is: {}--------------------------------------------",
                    eTaskStatus.getDesc());
            if (eTaskStatus == ETaskStatus.FAILED
                    || eTaskStatus == ETaskStatus.KILLED
                    || eTaskStatus == ETaskStatus.NOT_FOUND
                    || eTaskStatus == ETaskStatus.SUCCEEDED) {
                break;
            }

            Thread.sleep(3000);
        }

        if (eTaskStatus != null && eTaskStatus != ETaskStatus.SUCCEEDED) {
            LOG.error(
                    "------------------------------------current job status is: {}--------------------------------------------",
                    eTaskStatus.getDesc());
            System.exit(1);
        }
    }
}
